package com.netty.rpc.proxy;

import com.netty.rpc.client.NettyClient;
import com.netty.rpc.client.NettyClientHandler;
import com.netty.rpc.req.CalcRequest;
import com.netty.rpc.resp.Response;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class RpcClientProxy implements InvocationHandler {

    private static final Map<String, Channel> channelMap = new ConcurrentHashMap<>();

    private Map<String, CompletableFuture<Response>> completableFutureMap = new ConcurrentHashMap<>();

    public <T> T  getRpcClientProxy(Class<T> clazz){
        return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class<?>[]{clazz}, this);
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        Channel channel = null;
        CompletableFuture<Response> completableFuture = new CompletableFuture<>();
        completableFutureMap.put("1", completableFuture);
        if(channelMap.containsKey("CalcController")){
            channel =  channelMap.get("CalcController");
        }else{
            NettyClientHandler nettyClientHandler = new NettyClientHandler(completableFutureMap);
            channel = new NettyClient(nettyClientHandler).doConnect();
            channelMap.put("CalcController", channel);
        }
        CalcRequest request = new CalcRequest();
        request.setServiceName(method.getDeclaringClass().getSimpleName());
        request.setMethodName(method.getName());
        request.setParamsTypes(method.getParameterTypes());
        request.setParams(args);
        channel.writeAndFlush(request).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                log.info("client send message: success");
            } else {
                future.channel().close();
                completableFuture.completeExceptionally(future.cause());
                log.error("Send failed:", future.cause());
            }
        });

        return completableFuture.get();
    }
}
